package org.mq.mymq.broker.event.listener;

import com.alibaba.fastjson.JSON;
import io.netty.util.AttributeKey;
import org.mq.mymq.broker.cache.CommonCache;
import org.mq.mymq.broker.event.model.StartSyncEvent;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.dto.StartSyncRespDTO;
import org.mq.mymq.common.enums.BrokerResponseCode;
import org.mq.mymq.common.event.Listener;

import java.net.InetSocketAddress;

/**
 * @ClassName StartSyncListener
 * @Description 开启同步监听器
 * @Author jiarong_ye
 * @Date 2025/7/11 12:33
 * @Version 1.0
 */
public class StartSyncListener implements Listener<StartSyncEvent> {
    @Override
    public void onReceive(StartSyncEvent event) throws Exception {
        InetSocketAddress inetSocketAddress = (InetSocketAddress) event.getChannelHandlerContext().channel().remoteAddress();
        String reqId = inetSocketAddress.getAddress() + ":" + inetSocketAddress.getPort();
        event.getChannelHandlerContext().attr(AttributeKey.valueOf("reqId")).set(reqId);
        CommonCache. getSlaveChannelMap().put(reqId, event.getChannelHandlerContext());
        StartSyncRespDTO dto = new StartSyncRespDTO();
        dto.setSuccess(true);
        dto.setMsgId(event.getMsgId());
        TcpMsg respMsg = new TcpMsg(BrokerResponseCode.START_SYNC_SUCCESS.getCode(), JSON.toJSONBytes(dto));
        event.getChannelHandlerContext().writeAndFlush(respMsg);

    }
}
